package com.rainwen.zookeeper.client;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.api.CuratorListener;
import org.apache.curator.framework.api.CuratorWatcher;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.data.Stat;
import org.junit.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Random;
import java.util.concurrent.CountDownLatch;

/**
 * <pre>
 * 错误：java.io.IOException: Xid out of order. Got Xid 8 with err -101 expected Xid 7 for a packet with details
 * CuratorFramework 与 Zookeeper 版本不匹配
 * </pre>
 * Created by rain.wen on 2017/8/4.
 */
public class ZookeeperTest {

    public static final Logger logger = LoggerFactory.getLogger(ZookeeperTest.class);

    private static String zookeeperConnectionString = "192.168.10.2:2181";

    public static CuratorFramework client = null;

    public static String myPath = "/rainwen";

    @Before
    public void init(){
        client = ZkUtils.createClient(zookeeperConnectionString);
    }

    @Test
    public void printTest() throws Exception {
        ZkUtils.recursionAllPathAndData(client);
    }

    @Test
    public void createTest() throws Exception {
        Stat stat = client.checkExists().forPath(myPath);
        if(stat == null) {
            //不存在,创建
            client.create().forPath(myPath);
        }
        //不存在,创建
        client.create().withMode(CreateMode.PERSISTENT_SEQUENTIAL).forPath(myPath + "_SEQ");
    }

    @Test
    public void createSubTest() throws Exception {
        client.create().withMode(CreateMode.PERSISTENT).forPath(myPath + "/child" + new Random().nextInt(1000));
    }

    @Test
    public void createSequentialTest() throws Exception {
        client.create().withMode(CreateMode.PERSISTENT_SEQUENTIAL).forPath(myPath + "_SEQ");
    }

    @Test
    public void getTest() throws Exception {
        PrintUtils.printJson(client.getChildren().forPath(myPath + "_SEQ"));
    }

    @Test
    public void setDataTest() throws Exception {
        client.setData().forPath(myPath, "test".getBytes());
    }

    @Test
    public void updateDataTest() throws Exception {
        client.setData().forPath(myPath, "test2".getBytes());
    }

    @Test
    public void getDataTest() throws Exception {
        System.out.println(new String(client.getData().forPath(myPath)));
    }


    @Test
    public void getChildren() throws Exception {
        PrintUtils.printJson(client.getChildren().forPath(myPath));
    }

    @Test
    public void setDataAsync() throws Exception {
        // this is one method of getting event/async notifications
        CuratorListener listener = new CuratorListener() {
            public void eventReceived(CuratorFramework client, CuratorEvent event) throws Exception {
                // examine event for details
                logger.info("==> " + PrintUtils.getJson(event));
            }
        };
        client.getCuratorListenable().addListener(listener);
        // set data for the given node asynchronously. The completion
        // notification
        // is done via the CuratorListener.
        byte[] payload = "test2".getBytes();
        client.setData().inBackground().forPath(myPath, payload);
        logger.info("==> set data end");
        new CountDownLatch(1).await();
    }

    /**
     * 设置Data异步返回结果
     * @throws Exception
     */
    @Test
    public void setDataAsyncWithCallbackTest() throws Exception {
        BackgroundCallback callback = new BackgroundCallback() {
            public void processResult(CuratorFramework client, CuratorEvent event) throws Exception {
                logger.info("==> " + PrintUtils.getJson(event));
            }
        };
        byte[] payload = "newData".getBytes();
        // this is another method of getting notification of an async completion
        client.setData().inBackground(callback).forPath(myPath, payload);
        logger.info("==> set data end");
        new CountDownLatch(1).await();
    }

    /**
     * 永久的监听节点改变
     * zookeeper 默认的监听方式是一次性的，所以要循环监听
     * @throws Exception
     */
    @Test
    public void permanentListenerTest() throws Exception {
        CuratorWatcher wc = new CuratorWatcher() {
            /**
             * Same as {@link Watcher#process(WatchedEvent)}. If an exception
             * is thrown, Curator will log it
             *
             * @param event the event
             * @throws Exception any exceptions to log
             */
            public void process(WatchedEvent event) throws Exception {
                logger.info("==> {}", PrintUtils.getJson(event));
            }
        };
        while(true)
        {
            //当前节点数据改变
            client.checkExists().usingWatcher(wc).forPath(myPath);
            //子节点改变
            client.getChildren().usingWatcher(wc).forPath(myPath);
        }
    }
}
